package com.hhf.rrd.dataset;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;

/**
 * 计数任务
 *
 * @author huanghaifeng15
 * @date 2022/2/11 11:42
 **/
public class CounterApp {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String> dataSource = env.fromElements("hadoop", "spark", "flink", "spark");

        MapOperator<String, String> mapOperator = dataSource.map(new RichMapFunction<String, String>() {

            private LongCounter counter = new LongCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                getRuntimeContext().addAccumulator("ele-counter",counter);
            }

            @Override
            public String map(String value) throws Exception {
                // 可以对不符合规则的数据进行清洗
                counter.add(1);
                return value;
            }
        });

        mapOperator.writeAsText("data/out", FileSystem.WriteMode.OVERWRITE).setParallelism(2);
        JobExecutionResult executionResult = env.execute();
        Object counter = executionResult.getAccumulatorResult("ele-counter");
        System.out.println("counter --> " + counter);
    }
}
